package day02;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.FlatMapFunction;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.PairFunction;
import scala.Tuple2;

import java.lang.reflect.Array;
import java.util.Arrays;
import java.util.Iterator;

public class JavaWordCount {
    public static void main(String[] args) {
        //模块代码:获取Spark配置，然后给本应用程序设置一个名字，并且为本地测试执行。
        SparkConf conf = new SparkConf();
        conf.setAppName("JavaWordCount");
        conf.setMaster("local[2]"); //如果是集群运行就不用加这个setMaset("local[??]")
        //用conf配置获取对应的context。这里为获取JavaSparkContext
        JavaSparkContext jsp = new JavaSparkContext(conf);  //获取javaspark的上下文context，用来读取数据的

        //获取数据
        JavaRDD<String> lines = jsp.textFile("hdfs://master:9000/Wordcount3/input/one.txt");
        //切分 flatMap()
        JavaRDD<String> words = lines.flatMap(new FlatMapFunction<String, String>() {
            @Override
            public Iterator<String> call(String s) throws Exception {
                //方法返回值是Iterator，所以要用Arrays.asList(..)先变成List然后再用.iterator()变成可迭代的Iterator
                return Arrays.asList(s.split(" ")).iterator();
            }
        });
        //把单词生成元祖  mapToPair
        JavaPairRDD<String, Integer> tup = words.mapToPair(new PairFunction<String, String, Integer>() {
            @Override
            public Tuple2<String, Integer> call(String s) throws Exception {
                return new Tuple2<String, Integer>(s,1);
            }
        });
        //聚合
        JavaPairRDD<String, Integer> aggr = tup.reduceByKey(new Function2<Integer, Integer, Integer>() {
            @Override
            public Integer call(Integer v1, Integer v2) throws Exception {
                return v1 + v2;
            }
        });
        //排序，spark并没有javaAPI中提供sortBy()算子，所以我们可以把key-value对调一下然后用soryByKey()排序
        JavaPairRDD<Integer, String> swaped = aggr.mapToPair(new PairFunction<Tuple2<String, Integer>, Integer, String>() {
            @Override
            public Tuple2<Integer, String> call(Tuple2<String, Integer> tup) throws Exception {
                return tup.swap();
            }
        });
        JavaPairRDD<Integer, String> sorted = swaped.sortByKey(false);
        //排序之后，再次将key-value对调
        JavaPairRDD<String, Integer> res = sorted.mapToPair(new PairFunction<Tuple2<Integer, String>, String, Integer>() {
            @Override
            public Tuple2<String, Integer> call(Tuple2<Integer, String> sup) throws Exception {
                return sup.swap();
            }
        });
        //用collect()在控制台打印输出排序后的结果
        System.out.print(res.collect());
        //关闭资源
        jsp.stop();
    }


}
